/*
Copyright 2010 Bryce Alcock

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.com

 */

package com.ecocrypt.proxy.forwarder;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.net.Socket;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 * @author Bryce.Alcock
 */
public class EngineSocket implements Runnable, Serializable {
    //  TODO Dynamic-Properties
    public static final int SPIN_NANOS = 10;
    private static final int SPIN_MILLIS = 1;


    //-------------------------------------------------------------------------
    //  Client Service Request attribute included for purpose of marking
    //  The dirty flag.  This would not normally be required, but we want to
    //  make sure that we update on both sides of the socket.
    transient ClientServiceRequest csr ;

    //-------------------------------------------------------------------------
    //
    //   Active Runtime Attributes.
    //   These attributes of the Engine socket are transient in nature,
    //   they will not be de-serialized for post analysis.
    //--------------------------------------------------------------------------
    final transient LinkedList<Byte> sendBuffer = new LinkedList<Byte>();
    final transient LinkedList<Byte> recieveBuffer = new LinkedList<Byte>();
    transient Socket engineSocket;
    transient InputStream engineSocketInputStream;
    transient OutputStream engineSocketOutputStream;
    transient Forwarder forwarder;


    
    
    //--------------------------------------------------------------------------
    //
    //   Attributes and characteristics that will be stored for analysis of
    //   This particular client request.
    //--------------------------------------------------------------------------
    Integer serviceId;
    transient boolean isAlive = true;

    AtomicInteger bytesFromEngine = new AtomicInteger(0);
    AtomicInteger bytesToEngine = new AtomicInteger(0);


    //--------------------------------------------------------------------------
    //
    //   MEMORY LEAK POTENTIAL:
    //   The following pieces of data could cause a memory leak, and therefore
    //   Must be dealt with if the size of the data grows.
    //
    //--------------------------------------------------------------------------
    StringBuilder sentData = new StringBuilder();
    StringBuilder recieveData = new StringBuilder();
    LinkedBlockingQueue<TimingPill> sendTimes = new LinkedBlockingQueue<TimingPill>(500);
    LinkedBlockingQueue<TimingPill> recieveTimes = new LinkedBlockingQueue<TimingPill>(500);
    //
    //  End Memory Leak Potential Elements.
    //--------------------------------------------------------------------------
    LinkedList<String> shutdownMessage = new LinkedList<String>();

    private long startTime=System.currentTimeMillis();
    private long endTime=-1;
    private int localPort;
    private long lastByteReceived = 1;

    public EngineSocket(Forwarder f, Integer sid, ClientServiceRequest req)  throws IOException  {

	csr= req;
	serviceId = sid;
	forwarder = f;
	try {

	    engineSocket = new Socket(f.remoteMachine, f.remotePort);
            engineSocket.setKeepAlive(ForwardProxy.engineKeepAlive);
            
	    engineSocketInputStream = engineSocket.getInputStream();
	    engineSocketOutputStream = engineSocket.getOutputStream();
            localPort = engineSocket.getLocalPort();

	} catch (IOException ex) {
	    System.out.println("Unable to to Connect Engine Socket Rethrowing : " + ex.getMessage());
	    shutdownMessage.add("Unable to to Connect Engine Socket Rethrowing : " + ex.getMessage());
	    throw ex;
	}


    }

    public void startServiceThreads() {
	final String engineSocketInputThreadFormatString =
		"[%d]:[%d] Engine Input Thread %s";
	final String engineSocketOutputThreadFormatString =
		"[%d]:[%d] Engine Output Thread %s";

	ReaderClass rc = new ReaderClass();
	rc.setName(String.format(engineSocketInputThreadFormatString, serviceId, engineSocket.getLocalPort(),forwarder));
	rc.start();

	Thread senderThread = new Thread(this, String.format(engineSocketOutputThreadFormatString, serviceId, engineSocket.getLocalPort(),forwarder));
	senderThread.start();

    }

    private class ReaderClass extends Thread {

	@Override
	public void run() {

	    // 8 KB Buffer.
	    byte[] buf = new byte[1024 * 8];
	    int bytesIn = 0;
	    while (isAlive) {

		try {
		    if ((bytesIn = engineSocketInputStream.read(buf)) != -1) {
                        lastByteReceived=System.currentTimeMillis();

			// Timing Pill for bytes from engine.
			TimingPill tp = new TimingPill(bytesFromEngine.addAndGet(bytesIn));
			if(!recieveTimes.offer(tp)){
			    //  Memory Protection.
			    //  TODO: make this persistence work so that we don't
			    //        loose data if the number of pills exceeds 500.
			    recieveTimes.take();
			    recieveTimes.offer(tp);
			}
			forwarder.totalBytesFromEngine.addAndGet(bytesIn);


			synchronized (recieveBuffer) {
			    for (int i = 0; i < bytesIn; ++i) {
				recieveBuffer.add(new Byte(buf[i]));
			    }
			}

			if(ForwardProxy.saveReceiveData){
			    // Append the recieve Data if we have indicated
			    // that it is to be stored with the request.
                            char[] printChars = new char[bytesIn];
                            for(int i=0;i<bytesIn;++i){
                                //  If a char is less than ASCII 32, then replace with '~' (ASCII 126)
                                printChars[i]=(buf[i]<32&& buf[i]!=10 & buf[i]!=13)?(char)126:(char)buf[i];
                            }
			    if(recieveData.length()>1024000){
				// We have to remove some data.
				// TODO: Persist this Data.
				//       In order to prevent threads from blocking,
				//       a copy on write style persistance should be used.
				recieveData.setLength(0);
			    }
			    recieveData.append(printChars);
			}
			csr.dirty=true;

		    } else {
			//  This represents the NORMAL ending.
			//  We will close this socket down.
			//  Mark the conversation as good.
			isAlive = false;
			shutDown();
			csr.dirty=true;
		        shutdownMessage.add("EngineSocketInputStream -> End Of Stream");

		    }
		} catch (Throwable t) {
		    System.out.println("Throwable Event Causing Engine Socket to disconnect: " + t.getMessage());
		    t.printStackTrace(System.out);
		    isAlive = false;
		    shutdownMessage.add("EngineSocketInputStream -> EXCEPTION"+t.getMessage());
		    shutDown();
		    csr.dirty=true;
		    //  shut the socket down.
		}
	    }
	    shutdownMessage.add("EngineSocketInputStream -> isAlive false");
	}
    }

    @Override
    @SuppressWarnings("SleepWhileHoldingLock")
    public void run() {
	// This Thread will do any writting to the socket...
	while (isAlive) {
	    try {
		if (!sendBuffer.isEmpty()) {
		    int sendCount = 0;
		    byte[] bytesToSend;
		    synchronized (sendBuffer) {
			sendCount = sendBuffer.size();
			bytesToSend = new byte[sendCount];
			for (int i = 0; i < sendCount; ++i) {
			    bytesToSend[i] = sendBuffer.remove();
			}
		    }
		    try {
			
			//  now if sendCount>0, we will send the data down the socket.
			engineSocketOutputStream.write(bytesToSend);

			if(ForwardProxy.saveSendData){
                            String s = new String(bytesToSend);
                            //Ticket 6. Remove all non-ascii chars from the print stream.
                            //          This will not affect what is sent down the line.
                            //          all chars less than 32 will be replaced with 126 '~'.
                            for(int i=0;i<bytesToSend.length;++i){
                                if(bytesToSend[i]<32&& bytesToSend[i]!=10 & bytesToSend[i]!=13){
                                    //  ASCII 126 is the '~' char.
                                    bytesToSend[i]=126;
                                }
                            }
			    if(sentData.length()>1024000){
				// Remove potential Memory Leak.
				// TODO:  This should be persisted somewhere.
				//        a Copy on write persistence strategy
				//        should be employeed to prevent this thread from blocking.
				sentData.setLength(0);

			    }
			    sentData.append(s);
			}

			//  Record Bytes sent and place in a timing pill for analysis.
			TimingPill tp = new TimingPill(bytesToEngine.addAndGet(sendCount));
			if(!sendTimes.offer(tp)){
			    // This will need to be dealt with some other way.
			    // Currently to prevent the proxy memory leak, we will just
			    // remove the element from the queue.
			    sendTimes.take();
			    if(!sendTimes.offer(tp)){
				System.out.println("[WARNING] Unable to add Timing Pill to sendTimes bounded queue. sendTimes.poll() did not help. EngineSocket");
			    }
			}
			forwarder.totalBytesToEngine.addAndGet(sendCount);

		    } catch (IOException ex) {

			shutdownMessage.add("Failure to Write to Engine: " +ex.getMessage());
			// if we get an exception here, we will kill the full duplex socket.
			isAlive = false;
			// shut the socket down.
			shutDown();

		    }finally{
			csr.dirty=true;
		    }
		} else {
		    try {
			Thread.sleep(SPIN_MILLIS, SPIN_NANOS);
		    } catch (InterruptedException ex) {
			ex.printStackTrace(System.out);
		    }
		}
	    } catch (Throwable t) {
		shutdownMessage.add("Excpetion with EngineOutput Thread: " +t.getMessage());
		isAlive = false;
		shutDown();
		csr.dirty=true;
	    }
	}
	shutdownMessage.add("EnigneOuptputThread -> isAlive false");
    }

    public void disconnectEngine() {
	shutDown();
    }

    private void shutDown() {

	try {
	    isAlive = false;

	    try {
		engineSocketOutputStream.close();
	    } catch (Exception e) {
		System.out.println("Exception while shuting down Socket Stream -> OutputStream " + e.getMessage());
	    }

	    try {
		engineSocketInputStream.close();
	    } catch (Exception e) {
		System.out.println("Exception while shuting down Socket Stream -> InputStream " + e.getMessage());
	    }

	    if(engineSocket!=null){
	    if (!engineSocket.isClosed()) {

		engineSocket.close();

	    }
	    }

	    engineSocket = null;

	} catch (IOException ex) {
	    ex.printStackTrace(System.out);
	}finally{
	    csr.dirty=true;
	    endTime = System.currentTimeMillis();
	}

    }

    public void addSend(List<Byte> sendData) {
	synchronized (sendBuffer) {
	    sendBuffer.addAll(sendData);
	}
    }

    public List<Byte> getSocketData() {
	synchronized (recieveBuffer) {
	    LinkedList<Byte> returnList = new LinkedList<Byte>(recieveBuffer);
	    recieveBuffer.clear();
	    return returnList;
	}
    }

    public int getRecieveBufferSize() {
	synchronized (recieveBuffer) {
	    return recieveBuffer.size();
	}
    }

    
   /**
     * @return the lastByteReceived
     */
    public long getLastByteReceived() {
        return lastByteReceived;
    }

    public LinkedList<Byte> getRecieveBufferData() {
	LinkedList<Byte> sendData = new LinkedList<Byte>();
	synchronized (recieveBuffer) {
	    sendData.addAll(recieveBuffer);
	    recieveBuffer.clear();
	}
	return sendData;
    }

    public boolean isEngineDone(){
	return !isAlive;
    }

    public int getLocalPort(){
        return localPort;
    }

        @Override
    public String toString(){
	StringBuilder sb = new StringBuilder();
	sb.append("ServiceId:").append(serviceId).append("\n");
	sb.append("ENGINE LOCAL PORT:").append(localPort).append("\n");
	sb.append("StartTime:").append(new Date(startTime)).append("\n");
	sb.append("Durration").append((endTime<startTime) ? new Long(System.currentTimeMillis()-startTime): new Long(endTime-startTime)).append("\n");
	sb.append("Bytes From Engine:").append(bytesFromEngine.get()).append("\n");
	sb.append("Bytes To Engine:").append(bytesToEngine.get()).append("\n");
	sb.append("DataSent:").append(sentData).append("\n");
	sb.append("DataRecieved:").append(recieveData).append("\n");
	sb.append("Send Timing Pills:\n");
	for (TimingPill timingPill : sendTimes) {
	    sb.append("\t").append(timingPill.toString()).append("\n");
	}
	sb.append("Recieve Timing Pills:\n");
	for (TimingPill timingPill : recieveTimes) {
	    sb.append("\t").append(timingPill.toString()).append("\n");
	}
	sb.append("Socket ShutDown Reason:\n");
	for (String reason : shutdownMessage) {
	    sb.append("\t").append(reason).append("\n");
	}
	return sb.toString();
    }
}
